package com.cqcis.uip.base.framework.core.impl;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.BeforeSet;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.Service;
import com.cqcis.uip.base.framework.core.ThreadPoolConfig;
import com.cqcis.uip.base.framework.core.impl.AbstractTaskContext.WriteRecord;
import com.cqcis.uip.base.framework.handlers.ClientBlockService;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;

/**
 * SOCKET客户端同步阻塞实现
 * 
 * @author huangmy
 * @date   2008-8-26
 */
public abstract class AbstractClientTaskContextAdaptor<T> implements Service {
	private final static String FIRST_SEND_DISPATCH_KEY = "FIRST_SEND_DISPATCH_KEY_" + AbstractClientTaskContextAdaptor.class.getName();
	private final static String SHEDULEFINISHQUEUE = "SHEDULEFINISHQUEUE_" + AbstractClientTaskContextAdaptor.class.getName();
	private final static String SCHEDULE_END = "SCHEDULE_END"; // 执行完毕
	private final static String SCHEDULE_WRITE = "SCHEDULE_WRITE"; // 执行完毕		
	private ClientBlockService clientBlockService; // 阻塞实现的服务
	private ThreadPoolExecutor dispatchTaskExecutor; // task分派线程池
	private RegisteredServiceContext serviceContext; // 服务上下文
		
	public void start(RegisteredServiceContext serviceContext) throws Exception {
		this.serviceContext = serviceContext;
		ThreadPoolConfig tpConfig = serviceContext.getRegisteredService().getDefaultThreadPoolConfig();
		this.dispatchTaskExecutor = LogServiceThreadPoolExecutor.getLogExecutor(tpConfig.getCoreSize(), tpConfig.getMaxSize(),
				tpConfig.getKeepAliveTime(), 
				serviceContext.getLogService(), serviceContext.getRegisteredService().getLogConfiguration(),
				"Client Block Service " + serviceContext.getRegisteredService().getName());
		
		if (this.clientBlockService instanceof Service) {
			((Service)this.clientBlockService).start(serviceContext);
		}
	}
	
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		if (this.clientBlockService instanceof Service) {
			((Service)this.clientBlockService).stop(serviceContext);
		}
		
		if (this.dispatchTaskExecutor != null) {
			this.dispatchTaskExecutor.shutdown();
		}
	}
	
	/**
	 * 接收到的数据, 供 doMessageReceived() 方法使用,
	 * 不能在同一线程中顺序执行 doWirte 和该方法
	 * 
	 * @param message
	 * @param task
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	protected void messageSyncReceived(T message, Task task) throws Exception {		
		((AbstractTaskContext<T>)(task.getTaskContext())).offerMessage(message); // 添加到队列中
		Object record = getSheduleFinishQueueFor(task).poll();
		if (record == null) {
			throw new UIPServiceException("任务执行队列为空.");
		}
						
		if ((record instanceof Throwable) || SCHEDULE_WRITE.equals(record)) {
			task.setExecuteDefault(false);// 返回null，不执行默认异常操作
			task.setGenerateResult(false);
			
			return;
		}
		
		if (SCHEDULE_END.equals(record)) {
			getSheduleFinishQueueFor(task).offer(SCHEDULE_END);			
			return;
		}
	}
	
	/**
	 * 正常结束后
	 * 
	 */
	protected void afterNormalFinished(Task task) {
		// for subclass override
	}
	
	/**
	 * 是否完成
	 * 
	 * @param task
	 * @return
	 */
	public boolean isFinished(Task task) {
		Object record = getSheduleFinishQueueFor(task).peek();
		if (SCHEDULE_END.equals(record)) {
			return true;
		}
		
		return false;
	}
	
	/**
	 * 创建task上下文
	 * 
	 * @param serviceContext
	 * @param task
	 * @return
	 */
	protected abstract AbstractTaskContext<T> createTaskContext(RegisteredServiceContext serviceContext, Task task);

	/**
	 * 提交队列
	 * 
	 * @param task
	 * @return
	 */
	@SuppressWarnings("unchecked")
	private ConcurrentTaskQueue<Object> getSheduleFinishQueueFor(Task task) {
		return (ConcurrentTaskQueue<Object>)task.getAttribute(SHEDULEFINISHQUEUE);
	}
	
	/**
	 * 写控制, 控制第一次写数据，是否唤醒接受数据操作，
	 * 一般要求的执行写write操作和doMessageReceived操作在同一线程中时，返回false;
	 * 不在同一线程中时，返回true.
	 * 
	 * @return
	 */
	private boolean isWriteControlEnable() {
		return true;
	}
	
	/**
	 * 数据发送, 指连接成功，开始发送数据时调用，供 doMessageSend()方法使用
	 * 
	 * @param task
	 * @return
	 * @throws Exception
	 */
	protected T messageSyncSend(final Task task) throws Exception {
		task.setExecuteDefault(false);// 返回null，不执行默认异常操作
		
		Object key = task.getAttribute(FIRST_SEND_DISPATCH_KEY);
		if (key != null) {
			return null;
		}
		
		task.setAttribute(FIRST_SEND_DISPATCH_KEY, new Object());
		final AbstractTaskContext<T> taskContext = createTaskContext(this.serviceContext, task);// contxt
		taskContext.setWriteRecord(new WriteRecord() {
			private boolean firstWrited = true; // 是否第一次写
			public void write(Object o) {				
				synchronized (this) {
					if (firstWrited && isWriteControlEnable()) {
						firstWrited = false;
						return;// 第一次写，写不返回
					}
				}
				getSheduleFinishQueueFor(task).offer(SCHEDULE_WRITE);
			}
		});
		
		task.setAttribute(SHEDULEFINISHQUEUE, new ConcurrentTaskQueue<Object>());// 执行通信队列
		task.setTaskContext(taskContext);
		task.beforeSetAgain(new BeforeSet() {
			public void before() {
				throw new UIPServiceException("同步方式阻塞实现, BLOCKING不允许使用setAgain操作.");
			}
		});
		
		task.setAttribute(FrameworkConstants.TASK_BLOCKINNG_PROCESS_SYNC, new CountDownLatch(1));
		
		this.dispatchTaskExecutor.execute(new Runnable() {
			public void run() {
				try {
					CountDownLatch countLatch = (CountDownLatch)task.getAttribute(FrameworkConstants.TASK_BLOCKINNG_PROCESS_SYNC);
					if (countLatch != null) {
						countLatch.await();
					}
					taskContext.setContextRelaThread(Thread.currentThread());// 线程
					clientBlockService.schedule(task);
					getSheduleFinishQueueFor(task).offer(SCHEDULE_END);
					afterNormalFinished(task);
				} catch (Throwable t) {
					getSheduleFinishQueueFor(task).offer(t);
					serviceContext.catchException(task, t);					
				} finally {
					taskContext.setContextRelaThread(null);
				}
			}
		});
				
		notifyCommitEnd(task);
		
		return null;
	}
	
	/**
	 * 提交结束
	 * 
	 * @param task
	 */
	protected void notifyCommitEnd(Task task) {
		CountDownLatch countLatch = (CountDownLatch)task.removeAttribute(FrameworkConstants.TASK_BLOCKINNG_PROCESS_SYNC);
		if (countLatch != null) {
			countLatch.countDown();
		}
	}
    	
	public void setClientBlockService(ClientBlockService clientBlockService) {
		this.clientBlockService = clientBlockService;
	}
}
